package pro.fraud_detect.config;

import org.apache.flink.api.java.utils.ParameterTool;

import java.util.Arrays;
import java.util.List;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/2/22
 **/
public class Parameters {

    private final ParameterTool parameterTool;

    public Parameters(ParameterTool parameterTool) {
        this.parameterTool = parameterTool;
    }

    <T> T getOrdefault(Param<T> param){
        if(!parameterTool.has(param.getName())){
            return param.getDefaultValue();
        }

        Object value;

        if(param.getType()==Integer.class){
            value = parameterTool.getInt(param.getName());
        }else if(param.getType()==Long.class){
            value = parameterTool.getLong(param.getName());
        }else if(param.getType()==Double.class){
            value = parameterTool.getDouble(param.getName());
        }else if(param.getType()==Boolean.class){
            value = parameterTool.getBoolean(param.getName());
        }else{
            value = parameterTool.get(param.getName());
        }

        return param.getType().cast(value);

    }
    // Kafka:
    public static final Param<String> KAFKA_HOST = Param.string("kafka-host", "doitedu01");
    public static final Param<Integer> KAFKA_PORT = Param.integer("kafka-port", 9092);

    public static final Param<String> DATA_TOPIC = Param.string("data-topic", "livetransactions");
    public static final Param<String> ALERTS_TOPIC = Param.string("alerts-topic", "alerts");
    public static final Param<String> RULES_TOPIC = Param.string("rules-topic", "rules");
    public static final Param<String> LATENCY_TOPIC = Param.string("latency-topic", "latency");
    public static final Param<String> RULES_EXPORT_TOPIC =
            Param.string("current-rules-topic", "current-rules");

    public static final Param<String> OFFSET = Param.string("offset", "earliest");

    // Socket
    public static final Param<Integer> SOCKET_PORT = Param.integer("pubsub-rules-export", 9999);

    // General:
    //    source/sink types: kafka / pubsub / socket / STDOUT
    public static final Param<String> RULES_SOURCE_TYPE = Param.string("rules-source-type", "KAFKA");  // 原SOCKET
    public static final Param<String> DATA_SOURCE_TYPE = Param.string("data-source-type", "KAFKA"); // 或者GENERATOR来生成
    public static final Param<String> ALERTS_SINK_TYPE = Param.string("alerts-sink-type", "KAFKA");  // 原STDOUT
    public static final Param<String> LATENCY_SINK_TYPE = Param.string("latency-sink-type", "KAFKA");  // 原STDOUT
    public static final Param<String> RULES_EXPORT_SINK_TYPE = Param.string("rules-export-sink-type", "KAFKA");  // 原STDOUT

    public static final Param<Integer> RECORDS_PER_SECOND = Param.integer("records-per-second", 2);
    public static final Param<Boolean> LOCAL_EXECUTION = Param.bool("local", true);  // 原 false
    public static final Param<Integer> SOURCE_PARALLELISM = Param.integer("source-parallelism", 2);
    public static final Param<Boolean> ENABLE_CHECKPOINTS = Param.bool("checkpoints", false);
    public static final Param<Integer> CHECKPOINT_INTERVAL =
            Param.integer("checkpoint-interval", 60_000_0);
    public static final Param<Integer> MIN_PAUSE_BETWEEN_CHECKPOINTS =
            Param.integer("min-pause-btwn-checkpoints", 10000);
    public static final Param<Integer> OUT_OF_ORDERNESS = Param.integer("out-of-orderdness", 500);

    //  List<Param> list = Arrays.asList(new String[]{"foo", "bar"});

    public static final List<Param<String>> STRING_PARAMS =
            Arrays.asList(
                    KAFKA_HOST,
                    DATA_TOPIC,
                    ALERTS_TOPIC,
                    RULES_TOPIC,
                    LATENCY_TOPIC,
                    RULES_EXPORT_TOPIC,
                    OFFSET,
                    RULES_SOURCE_TYPE,
                    DATA_SOURCE_TYPE,
                    ALERTS_SINK_TYPE,
                    LATENCY_SINK_TYPE,
                    RULES_EXPORT_SINK_TYPE);

    public static final List<Param<Integer>> INT_PARAMS =
            Arrays.asList(
                    KAFKA_PORT,
                    SOCKET_PORT,
                    RECORDS_PER_SECOND,
                    SOURCE_PARALLELISM,
                    CHECKPOINT_INTERVAL,
                    MIN_PAUSE_BETWEEN_CHECKPOINTS,
                    OUT_OF_ORDERNESS);

    public static final List<Param<Boolean>> BOOL_PARAMS =
            Arrays.asList(LOCAL_EXECUTION, ENABLE_CHECKPOINTS);

}
